package com.common;

import com.google.gson.Gson;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class KafkaProducer {

    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaProducer.class);

    private static final String PRODUCER_TYPE = "async";

    private static Producer<String, String> producer;

    static {
        getProducerConfig();
    }

    public static void getProducerConfig() {
        Properties props = new Properties();
        props.put(KafkaConstant.BROKERLIST,
                KafkaConfigUtils.getProps(KafkaConstant.BROKERLIST));
        props.put(KafkaConstant.SERIALIZERCLASS, KafkaConfigUtils.getProps(
                KafkaConstant.SERIALIZERCLASS, "kafka.serializer.StringEncoder"));
        props.put(KafkaConstant.REQUESTACKS,
                KafkaConfigUtils.getProps(KafkaConstant.REQUESTACKS, "1"));
        props.put(KafkaConstant.REQUESTTIMEOUTMS,
                KafkaConfigUtils.getProps(KafkaConstant.REQUESTTIMEOUTMS, "10000"));
        props.put(KafkaConstant.CODEC,
                KafkaConfigUtils.getProps(KafkaConstant.CODEC, "none"));
        // 消息发送类型，sync同步，async异步
        String producerType = KafkaConfigUtils.getProps(KafkaConstant.PRODUCERTYPE);
        if (StringUtils.equalsIgnoreCase(producerType, PRODUCER_TYPE)) {
            props.put(KafkaConstant.PRODUCERTYPE, producerType);
            props.put(KafkaConstant.QUEUEMAXMS,
                    KafkaConfigUtils.getProps(KafkaConstant.QUEUEMAXMS, "5000"));
            props.put(KafkaConstant.QUEUEMAXMESSAGE,
                    KafkaConfigUtils.getProps(KafkaConstant.QUEUEMAXMESSAGE, "10000"));
            props.put(KafkaConstant.QUEUETIMEOUTMS,
                    KafkaConfigUtils.getProps(KafkaConstant.QUEUETIMEOUTMS, "-1"));
            props.put(KafkaConstant.QUEUEBATCHNUM,
                    KafkaConfigUtils.getProps(KafkaConstant.QUEUEBATCHNUM, "200"));
        }

        producer = new Producer<String, String>(new ProducerConfig(props));
    }

    /**
     * 
     * 功能描述: <br>
     * 使用kafkatopic进行发送
     *
     * @param obj
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    public static void sendMessage(Object obj) {
        String jsonMsg = new Gson().toJson(obj);
        LOGGER.info("try to send data to bigdata:{}", jsonMsg);
        KeyedMessage<String, String> data = new KeyedMessage<String, String>(
                KafkaConfigUtils.getProps(KafkaConstant.TOPIC), jsonMsg);
        producer.send(data);
    }

    /**
     * 
     * 功能描述: <br>
     * 使用指定的kafkatopic进行发送
     *
     * @param jsonMsg
     * @param kafkaTopic
     * @see [相关类/方法](可选)
     * @since [产品/模块版本](可选)
     */
    public static void sendMessage(String jsonMsg, String kafkaTopic) {
        LOGGER.debug("try to send member sensitive data to bigdata:{}", jsonMsg);
        KeyedMessage<String, String> data = new KeyedMessage<String, String>(
                KafkaConfigUtils.getProps(kafkaTopic), jsonMsg);
        producer.send(data);
    }

}
